package com.example;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.logging.log4j.util.Supplier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.event.EventListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;

@SpringBootApplication
@EnableBinding(Kevin.class)
@RestController
public class App {

    @Autowired
    private ApplicationContext context;

    @Autowired
    private InteractiveQueryService interactiveQueryService;

    @GetMapping("/page")
    public String page(){
        List<String> names = Arrays.asList("mfisher", "dyser", "schacko", "abilan", "ozhurakousky", "grussell");
        List<String> pages = Arrays.asList("blog", "sitemap", "initializr", "news", "colophon", "about");
        String rPage = pages.get(new Random().nextInt(pages.size()));
        String rName = names.get(new Random().nextInt(names.size()));
        PageView pageView = new PageView(rName, rPage, Math.random() > .5 ? 10 : 1000);
        PageViewEvent pageViewEvent = new PageViewEvent(this, pageView);
        context.publishEvent(pageViewEvent);
        System.out.println(String.format("发送消息成功,name:%s ,page:%s",rName,rPage));
        return "success";
    }

    @GetMapping("/counts")
    public Map<String,Long> counts(){
        Map<String, Long> counts = new HashMap<>();
        ReadOnlyKeyValueStore<String, Long> keyValue = interactiveQueryService.getQueryableStore("k_page_view_count", QueryableStoreTypes.keyValueStore());
        KeyValueIterator<String, Long> all = keyValue.all();
        while (all.hasNext()){
            KeyValue<String, Long> v = all.next();
            counts.put(v.key,v.value);
        }
        return counts;
    }

    @Data
    public static class PageViewEvent extends ApplicationEvent {
        PageView pageView;

        public PageViewEvent(Object source, PageView pageView) {
            super(source);
            this.pageView=pageView;
        }
    }

    @Component
    public static class PageViewEventSink{
        @Autowired
        @Qualifier(Kevin.OUTPUT)
        MessageChannel channel;

        @EventListener(PageViewEvent.class)
        public void pageView(PageViewEvent event){
            System.out.println("接收到消息");
            channel.send(MessageBuilder
                    .withPayload(event.getPageView())
                    .setHeader(KafkaHeaders.MESSAGE_KEY,event.getPageView().getUserId().getBytes())
                    .build());
            System.out.println("发出成功!!!!");
        }

        @Bean
        public Function<KStream<String, PageView>,KStream<String, Long>>  process(){

            return p->p.filter((k, v)-> v.getDuration()>10)
                    .map((k, v)-> new KeyValue<>(v.getPage(),"0"))
                    .groupByKey(Grouped.with(Serdes.String(),Serdes.String()))
                    .count(Materialized.as("k_page_view_count"))
                    .toStream();
        }
    }

    @Bean
    public Consumer<KTable<String,Long>> consumer(){
        return e->e.toStream().foreach((k, v)->
                System.out.println(String.format("消费消息:key: %s ,value: %s",k,v)));
    }


    public static void main(String[] args) {
        SpringApplication.run(App.class, args);
    }
}
interface Kevin{

    String OUTPUT = "kout";

    @Output(OUTPUT)
    MessageChannel output();


}

@Data
@NoArgsConstructor
@AllArgsConstructor
class PageView {
    private String userId,page;
    private long duration;
}